package com.zhang.sql2;

import com.zhang.first.utils.ClickEvent;
import com.zhang.first.utils.ClickEventSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @title:
 * @author: zhang
 * @date: 2022/3/22 20:44
 */
public class TimeAndWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //读取数据源
        //TODO 1 创建表 (连接器表) 在创建表的ddl中直接定义时间属性
        String create = "create table clickTable (" +
                " name STRING," +
                " url STRING," +
                " ts BIGINT," +
                " et AS TO_TIMESTAMP( FROM_UNIXTIME(ts/1000) )," +
                " WATERMARK FOR et AS et - INTERVAL '1' SECOND " +
                " ) WITH ( " +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/click.txt'," +
                " 'format' = 'csv'" +
                " )";
        tableEnv.executeSql(create);

        //TODO 2 在流创建流的时候指定时间属性
        SingleOutputStreamOperator<ClickEvent> clickDS = env
                .addSource(new ClickEventSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ClickEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                                    @Override
                                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        Table fromDataStream = tableEnv.fromDataStream(clickDS, $("name"), $("url"), $("ts"), $("ts").rowtime().as("rt"));
        tableEnv.createTemporaryView("click",fromDataStream);
        //fromDataStream.printSchema();
        //Table sqlQuery = tableEnv.sqlQuery("select * from clickTable");
        //tableEnv.toDataStream(sqlQuery).print();

        //分组聚合
        Table aggTable = tableEnv.sqlQuery("select name,count(1) from clickTable group by name");
        //tableEnv.toChangelogStream(aggTable).print("agg");

        //分组窗口聚合
        Table windowTable = tableEnv.sqlQuery("" +
                " select name,count(1) as cnt," +
                " TUMBLE_END(et,INTERVAL '5' SECOND) as ent " +
                " from clickTable " +
                " group by " +
                " name," +
                " TUMBLE(et,INTERVAL '5' SECOND)");
        //tableEnv.toDataStream(windowTable).print("group window");

        //窗口聚合TVF
        //滚动窗口
        Table tvfTumbleTable = tableEnv.sqlQuery("" +
                " select name," +
                " count(1) as cnt," +
                " window_end ent," +
                " window_start stt" +
                " from TABLE(" +
                " TUMBLE( TABLE clickTable, " +
                " DESCRIPTOR(et), " +
                " INTERVAL '5' SECOND)) " +
                " group by name,window_start,window_end");
        //滑动窗口
        Table tvfHopTable = tableEnv.sqlQuery("" +
                " select name," +
                " count(1) as cnt," +
                " window_end ent," +
                " window_start stt" +
                " from TABLE(" +
                " HOP( TABLE clickTable, " +
                " DESCRIPTOR(et)," +
                " INTERVAL '5' SECOND, " +
                " INTERVAL '10' SECOND)) " +
                " group by name,window_start,window_end");

        //累积窗口
        Table tvfCumulateTable = tableEnv.sqlQuery("" +
                " select name," +
                " count(1) as cnt," +
                " window_end ent," +
                " window_start stt" +
                " from TABLE(" +
                " CUMULATE( TABLE clickTable, " +
                " DESCRIPTOR(et)," +
                " INTERVAL '5' SECOND, " +
                " INTERVAL '10' SECOND)) " +
                " group by name,window_start,window_end");
        tableEnv.toDataStream(tvfTumbleTable).print("tvfCumulate");

        //开窗（over）聚合
        Table overTable = tableEnv.sqlQuery("" +
                " select name," +
                " avg(ts) OVER(" +
                " PARTITION BY name " +
                " ORDER BY et " +
                " ROWS BETWEEN 3 PRECEDING AND CURRENT ROW)" +
                " AS avg_ts " +
                " from clickTable");
        //tableEnv.toDataStream(overTable).print("overTable");


        env.execute();
    }
}
